package com.cbz.learn.kafka;


import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.HashMap;
import java.util.Map;

/**
 * 测试kafka生产者
 */
public class KafkaProducerTest {

    public static void main(String[] args) {
        Map<String, Object> config = new HashMap<>();
        // 设置kafka服务地址
        config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        // 对生产者的key value 做序列化配置
        config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        // 创建生产者对象
        KafkaProducer<String, String> producer = new KafkaProducer<>(config);

        // 持续发送十条数据
        for (int i = 0; i < 10; i++) {
            // 创建数据
            ProducerRecord<String, String> record = new ProducerRecord<>("testTopic", "key"  +i, "this is record" + i);
            producer.send(record);
        }

        // 发送完后需要关闭生产者对象
        producer.close();
    }
}
